package com.cssl.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.cssl.bean.VisitorStats;
import com.cssl.utils.ClickHouseUtil;
import com.cssl.utils.CommonUtils;
import com.cssl.utils.DateTimeUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple4;

import java.time.Duration;
import java.util.Date;
import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/17 9:19
 * @Desc: 访客主题宽表
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        //1. 获取执行环境
        Properties properties = CommonUtils.getProperties();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 读取kafka数据创建流
        String uvSourceTopic = properties.getProperty("kafka.topic.log.dwm.uv");
        DataStreamSource<String> uvDS = env.addSource(CommonUtils.getKafkaConsumer(uvSourceTopic));

        String ujSourceTopic = properties.getProperty("kafka.topic.log.dwm.jump");
        DataStreamSource<String> ujDS = env.addSource(CommonUtils.getKafkaConsumer(ujSourceTopic));

        String pvSourceTopic = properties.getProperty("kafka.topic.log.dwd.page");
        DataStreamSource<String> pvDS = env.addSource(CommonUtils.getKafkaConsumer(pvSourceTopic));

        //3. 将每个流处理成相同的数据类型
        //3.1 处理 uv 数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithUvDS = uvDS.map(line -> {
            JSONObject jsonObject = JSONObject.parseObject(line);
            //提取common字段
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats(
                    "",
                    "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L,
                    0L,
                    0L,
                    0L,
                    0L,
                    jsonObject.getLong("ts"));
        });

        //3.2 处理 uj 数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithUjDS = ujDS.map(line -> {
            JSONObject jsonObject = JSONObject.parseObject(line);
            //提取common字段
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats(
                    "",
                    "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L,
                    0L,
                    0L,
                    1L,
                    0L,
                    jsonObject.getLong("ts"));
        });

        //3.3 处理 pv 数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithPvDS = pvDS.map(line -> {
            JSONObject jsonObject = JSONObject.parseObject(line);
            //提取common字段
            JSONObject common = jsonObject.getJSONObject("common");
            //提取page字段
            JSONObject page = jsonObject.getJSONObject("page");

            String lastPageId = page.getString("last_page_id");
            long sv = 0L;
            if (lastPageId == null || lastPageId.length() <= 0) {
                sv = 1L;
            }
            return new VisitorStats(
                    "",
                    "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L,
                    1L,
                    sv,
                    0L,
                    page.getLong("during_time"),
                    jsonObject.getLong("ts"));
        });

        //4. union
        DataStream<VisitorStats> unionDS = visitorStatsWithUvDS.union(visitorStatsWithUjDS, visitorStatsWithPvDS);

        //5. 提起时间戳生成watermark
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWaterMakeDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(11)).withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
            @Override
            public long extractTimestamp(VisitorStats element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        //6. 按照维度信息分组
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> keyedDS = visitorStatsWithWaterMakeDS.keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(VisitorStats value) throws Exception {
                return new Tuple4<String, String, String, String>(
                        value.getAr(),
                        value.getCh(),
                        value.getIs_new(),
                        value.getVc());
            }
        });

        //7. 开窗聚合 10s的滚动窗口
        //注意: uj 数据需要10秒的计算计算时间才出结果
        WindowedStream<VisitorStats, Tuple4<String, String, String, String>, TimeWindow> windowedDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<VisitorStats> resultDS = windowedDS.reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats value1, VisitorStats value2) throws Exception {
                        //注意: 滑动窗口只能用 new 的方式,因为滑动窗口数据可能会被多个窗口使用
                        value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                        value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                        value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                        value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                        value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());
                        return value1;
                    }
                },
                new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<VisitorStats> input, Collector<VisitorStats> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        VisitorStats visitorStats = input.iterator().next();
                        visitorStats.setStt(DateTimeUtil.toYMDhms(new Date(start)));
                        visitorStats.setEdt(DateTimeUtil.toYMDhms(new Date(end)));

                        out.collect(visitorStats);
                    }
                }
        );

        //8. 将数据写入clickhouse
        resultDS.addSink(ClickHouseUtil.getSink("insert into visitor_stats_2021 values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        //9. 启动任务
        env.execute("VisitorStatsApp");
    }
}
